package dosgi.distribution.remote.impl;

import java.io.ByteArrayOutputStream;
import java.io.IOException;
import java.io.ObjectOutput;
import java.io.ObjectOutputStream;

import com.rabbitmq.client.AMQP.BasicProperties;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import com.rabbitmq.client.QueueingConsumer;

import dosgi.distribution.Activator;
import dosgi.distribution.RemoteInvocationArguments;
import dosgi.distribution.invoker.RemoteInvoker;
import dosgi.distribution.invoker.impl.RemoteInvokerImpl;
import dosgi.distribution.remote.RemoteInvocationServer;

public class RemoteInvocationServerImpl implements RemoteInvocationServer {

	private ConnectionFactory factory;
	private Connection connection;
	private Channel channel;
	private String queueName = Activator.QUEUE_NAME;
	private QueueingConsumer consumer;
	private RemoteInvoker invoker;

	public RemoteInvocationServerImpl() throws Exception {
		invoker = new RemoteInvokerImpl();
		
		factory = new ConnectionFactory();

		// TODO pobieranie konfiguracji z pliku!
		factory.setHost("nn.evizone.icsadl.agh.edu.pl");

		connection = factory.newConnection();
		channel = connection.createChannel();

		//TODO kolejka bedzie miala nazwe autowygenerowana
		channel.queueDeclare(queueName, false, false, false, null);

		int prefetchCount = 1;
		channel.basicQos(prefetchCount);

		consumer = new QueueingConsumer(channel);
		channel.basicConsume(queueName, false, consumer);
	}

	@Override
	public void run() {
		try {
			while (true) {
				Object response = null;
	
				QueueingConsumer.Delivery delivery = consumer.nextDelivery();
				BasicProperties props = delivery.getProperties();
				BasicProperties replyProps = new BasicProperties.Builder().correlationId(props.getCorrelationId()).build();
				
				RemoteInvocationArguments args = RemoteInvocationArguments.getFromByteArray(delivery.getBody());
				
				try {
					response = invoker.invoke(args.getClassName(), args.getMethodName(), args.getArgs(), args.getProperties());
				} catch (IllegalArgumentException e) {
					// TODO Auto-generated catch block
					e.printStackTrace();
					response = null;
				} finally {
					channel.basicPublish("", props.getReplyTo(), replyProps, toByteArray(response));
					channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false);
				}
			}
		} catch (Exception e) {
			e.printStackTrace();
		} finally {
			if (connection != null) {
				try {
					connection.close();
				} catch (Exception ignore) {
				}
			}
		}
	}
	
	@Override
	public void close() throws Exception {
		channel.close();
		connection.close();
	}
	
	private byte[] toByteArray(Object obj) throws IOException {
		ObjectOutput out;
		ByteArrayOutputStream bos = new ByteArrayOutputStream();
		out = new ObjectOutputStream(bos);
		out.writeObject(obj);
		out.close();
		
		return bos.toByteArray();
	}

}
